package com.luoy.json;

import com.alibaba.fastjson.JSON;
import org.apache.commons.lang3.StringUtils;

import java.io.File;
import java.io.FileInputStream;
import java.nio.charset.StandardCharsets;
import java.util.List;

/**
 * @author weiwei
 * @Title: JobUtils
 * @ProjectName sugon_flink_demo
 * @Description: TODO
 * @date 2020/7/31 11:24
 * <p>
 * {"job": {"content": [{"reader": {"parameter": {"topic": "kafka11", "groupId": "default",
 * "consumerSettings": {"bootstrap.servers": "0.0.0.1:9092", "auto.commit.interval.ms": "1000",
 * "auto.offset.reset": "latest"}, "column": ["user_id", "name", "age"], "compareColumn": "name"},
 * "name": "kafka11reader"}, "compare": {"name": "mysql", "parameter": {"username": "root",
 * "password": "Sugon_456", "connection": [{"jdbcUrl": "jdbc:mysql://10.0.13.226:3306/test?useSSL=false",
 * "table": ["cs"] } ], "column": ["name"] } }, "writer": {"parameter": {"timezone": "UTC", "topic": "kafka11",
 * "producerSettings": {"bootstrap.servers": "0.0.0.1:9092"}, "tableFields": ["id", "name"] },
 * "name": "kafka11writer"} } ] } }
 */
public class JobUtils {

    public static ConfigurationJson getJsonConfiguration(String jobConfig) {
        String content = JSON.parseObject(jobConfig).getJSONObject("job")
                .getJSONArray("content")
                .getJSONObject(0).toString();

        ConfigurationJson configuration = ConfigurationJson.from(content);

        return configuration;
    }


    public static String readJob(File file) {
        try (FileInputStream in = new FileInputStream(file);) {
            byte[] fileContent = new byte[(int) file.length()];
            in.read(fileContent);
            return new String(fileContent, StandardCharsets.UTF_8);
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }


    public static void main(String[] args) {

//        String json = "{\"job\": {\"content\": [{\"reader\": {\"parameter\": {\"topic\": \"kafka11\", \"groupId\": \"default\", \"servers\": \"0.0.0.1:9092\", \"interval\": \"1000\", \"reset\": \"latest\", \"column\": [\"user_id\", \"name\", \"age\"], \"compareColumn\": \"name\"}, \"name\": \"kafka11reader\"}, \"compare\": {\"name\": \"mysql\", \"parameter\": {\"username\": \"root\", \"password\": \"Sugon_456\", \"connection\": {\"jdbcUrl\": \"jdbc:mysql://10.0.13.226:3306/test?useSSL=false\", \"table\": \"cs\"}, \"column\": [\"name\"] } }, \"writer\": {\"parameter\": {\"topic\": \"kafka11\", \"servers\": \"0.0.0.1:9092\", \"column\": [\"id\", \"name\"] }, \"name\": \"kafka11writer\"} } ] } }";

//        ConfigurationJson configuration = getReaderConfiguration(json);

//        System.out.println("topic:" + configuration.getString("parameter.topic"));
//        System.out.println("servers:" + configuration.getString("parameter.servers"));
//        System.out.println("groupId:" + configuration.getString("parameter.groupId"));

//
//        List<Object> list = configuration.getList("parameter.column");
//
//
//        String sql = String.format("select %s from %s",
//                StringUtils.join(list.toArray(), ","), "cs");
//
//        System.out.println(sql);

        String json = "{\"job\": {\"content\": [{\"reader\": {\"parameter\": {\"topic\": \"test\", \"groupId\": \"default\", \"servers\": \"slave03:6667\", \"interval\": \"1000\", \"reset\": \"latest\", \"column\": [\"user_id\", \"name\", \"age\"], \"compareColumn\": \"name\"}, \"name\": \"kafka11reader\"}, \"compare\": {\"name\": \"mysql\", \"parameter\": {\"username\": \"root\", \"password\": \"Sugon_456\", \"connection\": {\"jdbcUrl\": \"jdbc:mysql://10.0.13.226:3306/test?useSSL=false\", \"table\": \"test_kafka\"}, \"column\": [\"name\"] } }, \"writer\": {\"parameter\": {\"topic\": \"ww_cs01\", \"servers\": \"slave03:6667\", \"column\": [{\"name\": \"created_date\", \"type\": \"STRING\"}, {\"name\": \"id\", \"type\": \"STRING\"}, {\"name\": \"name\", \"type\": \"blob\"} ] }, \"name\": \"kafka11writer\"} } ] } }";


        ConfigurationJson configurationJson = getJsonConfiguration(json);


//        System.out.println(configurationJson.getString("compare.parameter.connection.jdbcUrl"));
//        System.out.println(configurationJson.getString("compare.parameter.password"));
//        System.out.println(configurationJson.getString("compare.parameter.username"));
//        System.out.println(configurationJson.getString("compare.parameter.connection.table"));
        List<ConfigurationJson> listConfiguration = configurationJson.getListConfiguration("writer.parameter.column");

        for (ConfigurationJson configurationJson1 : listConfiguration){

            System.out.println(configurationJson1.getString("type"));
            System.out.println(configurationJson1.getString("name"));
        }
    }


}
